#ifndef MDS_H
#define MDS_H
#include <string>
#include <map>
#include <vector>
#include <atomic>
#include <list>
#include <mutex>
#include <condition_variable>

#include "dss_common.h"
#include "mds_op_type.h"
#include "mds_cache_type.h"
#include "Transaction.h"
#include "Lock.h"
//#include "FileOpQueue.h"
#include "MiRpc.h"
#include "kv/KeyValueDB.h"

//using namespace std;


enum {
	PREPARE = 0,
	ACTIVE,
	EXIT,
};

struct rpc_msg_info {
	uint16_t rpc_type;
	uint16_t client_id;
	uint16_t msg_type;
	uint32_t msg_id;
	uint32_t data_len;
	void *data;
	string uri;
	RpcResponseHandler cb;
	int retval;
	void *ret;
	bool done;
	mutex lock;
	condition_variable cond;
};


class MDS {
public:
	uint64_t	id;
	string		conf;
	uint32_t	state;	//TODO need lock
	string		uri;
	string		master_uri;
	map<uint64_t, string> cs_uri;
	uint64_t	nodeid;
	mutex			nodeid_lock;

/*************  rpc_server  ****************/
public:
	RdmaRpc   *rpc_server;
	void			*mem0;
	MemoryManager *mm0;
	list<struct RpcExecution *> rpc_server_rx_q;
	mutex rpc_server_rx_q_lock;
	list<struct RpcExecution *> rpc_server_tx_q;
	mutex rpc_server_tx_q_lock;
	pthread_t  rpc_server_th;

	static void *rpc_server_main(void *arg);
	static void file_op_service(void* self, void* ex, void* mds);


/*************  rpc_client  ****************/
public:
	RdmaRpc   *rpc_client;
	void      *mem1;
	MemoryManager *mm1;
	map<string, int> uri_to_session_id;
	list< struct rpc_msg_info *> rpc_client_tx_q;
	mutex rpc_client_tx_q_lock;
	pthread_t  rpc_client_th;

	static void *rpc_client_main(void *arg);

/*************  worker thread pool  ****************/
public:
	int pool_size;
	vector<pthread_t> worker_thread_pool;
	static void *worker_process(void *arg);
	
	void handle_mknod(struct FileOpMKNOD *op, struct dss_msg *m_resp);
	void handle_mkdir(struct FileOpMKDIR *op, struct dss_msg *m_resp);
	void handle_unlink(struct FileOpUNLINK *op, struct dss_msg *m_resp);
	void handle_rmdir(struct FileOpRMDIR *op, struct dss_msg *m_resp);
	void handle_rename(struct FileOpRENAME *op, struct dss_msg *m_resp);
	void handle_stat(struct FileOpSTAT *op, struct dss_msg *m_resp);
	void handle_chmod(struct FileOpCHMOD *op, struct dss_msg *m_resp);
	void handle_chown(struct FileOpCHOWN *op, struct dss_msg *m_resp);
	void handle_open(struct FileOpOPEN *op, struct dss_msg *m_resp);
	void handle_read(struct FileOpREAD *op, struct dss_msg *m_resp);
	void handle_write(struct FileOpWRITE *op, struct dss_msg *m_resp);
	void handle_truncate(struct FileOpTRUNCATE *op, struct dss_msg *m_resp);
	void handle_close(struct FileOpCLOSE *op, struct dss_msg *m_resp);
	void handle_fsync(struct FileOpFSYNC *op, struct dss_msg *m_resp);
	void handle_readdir(struct FileOpREADDIR *op, struct dss_msg *m_resp);
	void handle_statfs(struct FileOpSTATFS *op, struct dss_msg *m_resp);

	int create_chunk(struct chunk_id *chunkid);
	int delete_chunk(struct chunk_id *chunkid);

	static void chunk_create_cb(void *arg1, void *arg2, void *ctx);
	static void chunk_delete_cb(void *arg1, void *arg2, void *ctx);

	void correct(const char *old_path, char *new_path);
	bool get_ppath_name(const char *path, char *parent, char *name);
/*************  kv  ****************/
public:
	string dir_meta_db_type;
	string dir_meta_db_path;
	KeyValueDB	*dir_meta_db;
	string dir_data_db_type;
	string dir_data_db_path;
	KeyValueDB	*dir_data_db;
	string file_meta_db_type;
	string file_meta_db_path;
	KeyValueDB	*file_meta_db;
	string file_data_db_type;
	string file_data_db_path;
	KeyValueDB	*file_data_db;


/*************  transaction  ****************/
public:
	Transaction	transaction;

/*************  filesystem lock  ****************/
public:
	LockService locks;

/*************  cache  ****************/
public:
	map<uint64_t, struct CInode> cinodes;
	mutex cinodes_lock;

/*********************************/
public:
	MDS(uint64_t id_, string conf_) :
		id(id_),
		conf(conf_),
		state(PREPARE),
		nodeid(0),
		rpc_server(NULL),
		mem0(NULL),
		mm0(NULL),
		rpc_client(NULL),
		mem1(NULL),
		mm1(NULL),
		pool_size(1),
		dir_meta_db(NULL),
		dir_data_db(NULL),
		file_meta_db(NULL),
		file_data_db(NULL)
	{}

	~MDS(){}
	int parse_conf();
	void prepare();
	void run();
	void exit();


};

#endif
